package com.bigknow.appstore.common.rpc.client;

import static com.bigknow.org.objectweb.asm.Opcodes.ACC_PRIVATE;
import static com.bigknow.org.objectweb.asm.Opcodes.ACC_PUBLIC;
import static com.bigknow.org.objectweb.asm.Opcodes.ACC_SUPER;
import static com.bigknow.org.objectweb.asm.Opcodes.ACONST_NULL;
import static com.bigknow.org.objectweb.asm.Opcodes.ALOAD;
import static com.bigknow.org.objectweb.asm.Opcodes.ARETURN;
import static com.bigknow.org.objectweb.asm.Opcodes.ASTORE;
import static com.bigknow.org.objectweb.asm.Opcodes.ATHROW;
import static com.bigknow.org.objectweb.asm.Opcodes.CHECKCAST;
import static com.bigknow.org.objectweb.asm.Opcodes.DLOAD;
import static com.bigknow.org.objectweb.asm.Opcodes.DRETURN;
import static com.bigknow.org.objectweb.asm.Opcodes.DSTORE;
import static com.bigknow.org.objectweb.asm.Opcodes.DUP;
import static com.bigknow.org.objectweb.asm.Opcodes.FLOAD;
import static com.bigknow.org.objectweb.asm.Opcodes.FRETURN;
import static com.bigknow.org.objectweb.asm.Opcodes.FSTORE;
import static com.bigknow.org.objectweb.asm.Opcodes.GETFIELD;
import static com.bigknow.org.objectweb.asm.Opcodes.IFEQ;
import static com.bigknow.org.objectweb.asm.Opcodes.ILOAD;
import static com.bigknow.org.objectweb.asm.Opcodes.INVOKEINTERFACE;
import static com.bigknow.org.objectweb.asm.Opcodes.INVOKESPECIAL;
import static com.bigknow.org.objectweb.asm.Opcodes.INVOKESTATIC;
import static com.bigknow.org.objectweb.asm.Opcodes.INVOKEVIRTUAL;
import static com.bigknow.org.objectweb.asm.Opcodes.IRETURN;
import static com.bigknow.org.objectweb.asm.Opcodes.ISTORE;
import static com.bigknow.org.objectweb.asm.Opcodes.LLOAD;
import static com.bigknow.org.objectweb.asm.Opcodes.LRETURN;
import static com.bigknow.org.objectweb.asm.Opcodes.LSTORE;
import static com.bigknow.org.objectweb.asm.Opcodes.NEW;
import static com.bigknow.org.objectweb.asm.Opcodes.PUTFIELD;
import static com.bigknow.org.objectweb.asm.Opcodes.RETURN;
import static com.bigknow.org.objectweb.asm.Opcodes.V1_5;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.lang.reflect.GenericArrayType;
import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.security.KeyStore;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicLong;

import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocket;
import javax.net.ssl.TrustManagerFactory;

import com.bigknow.appstore.common.rpc.*;
import com.bigknow.net.minidev.json.JSONStyle;
import com.bigknow.net.minidev.json.JSONValue;

import com.bigknow.org.objectweb.asm.ClassWriter;
import com.bigknow.org.objectweb.asm.Label;
import com.bigknow.org.objectweb.asm.MethodVisitor;
import com.bigknow.org.objectweb.asm.Type;

public class RPCClient {

    private String serviceHost;

    private int servicePort;

    private String keystore;

    private volatile static boolean connected;

    private volatile boolean closed;

    private SSLSocket socket;

    private DataInputStream dataInput;

    private static DataOutputStream dataOutput;

    private static AtomicLong idGen = new AtomicLong(System.currentTimeMillis());

    private Map<Class<?>, Class<?>> requestProxyMap = new ConcurrentHashMap<Class<?>, Class<?>>();

    private Map<String, Map<String, RPCHandler>> rpcHandlerMap = new ConcurrentHashMap<String, Map<String, RPCHandler>>();

    private ExecutorService pool;

    private static Map<String, RPCResponse> responseMap = new HashMap<String, RPCResponse>();

    public void init(String serviceHost, int servicePort, String keystore, ExecutorService pool) {
        this.serviceHost = serviceHost;
        this.servicePort = servicePort;
        this.keystore = keystore;
        this.pool = pool;
    }

    public void connect() {
        if (!closed) {
            try {
                _connect();
            } catch (Exception e) {
                connected = false;
                createReconnetTask();
            }
        }
    }

    public void reConnect() {
        if (!closed) {
            try {
                _close();
                connect();
            } catch (Exception e) {
                connected = false;
                createReconnetTask();
            }
        }
    }

    private synchronized void _connect() throws Exception {
        if (socket != null) {
            _close();
        }
        SSLContext ctx = SSLContext.getInstance("SSL");

        KeyManagerFactory kmf = KeyManagerFactory.getInstance("SunX509");
        TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509");

        KeyStore ks = KeyStore.getInstance("JKS");

        ks.load(new FileInputStream(keystore), "atlas".toCharArray());

        kmf.init(ks, "atlas".toCharArray());
        tmf.init(ks);

        ctx.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
        socket = (SSLSocket) ctx.getSocketFactory().createSocket(serviceHost, servicePort);
        socket.setSoTimeout(20000);
        dataInput = new DataInputStream(new BufferedInputStream(socket.getInputStream()));
        dataOutput = new DataOutputStream(new BufferedOutputStream(socket.getOutputStream()));
        dataOutput.write(MuxAction.MAIN.getCode().getBytes());
        dataOutput.flush();
        connected = true;
        Reader reader = new Reader();
        reader.setName("RPCClient_reader");
        reader.start();
//        System.out.println(String.format("服务器已连接（%s:%s）", serviceHost, servicePort));
    }

    private synchronized void _close() {
        if (socket != null) {
            try {
                socket.close();
                socket = null;
            } catch (IOException e) {
            }
        }
        connected = false;
    }

    public boolean isConnected() {
        return connected;
    }

    public void close() {
        closed = true;
        _close();
    }

    private void createReconnetTask() {
        Constants.timer.schedule(new TimerTask() {
            public void run() {
                if (!connected) {
                    _close();
                    connect();
                }
            }
        }, 1000);
    }

    public <T> void registerRPCInterface(Class<T> interfaceClass, T impl) {
        if (!interfaceClass.isInterface()) {
            throw new RuntimeException(interfaceClass.getName() + " 不是接口");
        }
        Map<String, RPCHandler> methodHandlers = new HashMap<String, RPCHandler>();
        Map<String, List<String>> parmNameMap = MethodParamNamesScaner.getParamNames(interfaceClass);
        for (Method method : interfaceClass.getDeclaredMethods()) {
            createRequestClass(method, RPCUtil.getImplClassName(interfaceClass),
                    parmNameMap.get(MethodParamNamesScaner.getKey(method)));
            createResponseClass(method, RPCUtil.getImplClassName(interfaceClass));
            String requestClass = RPCUtil.getMethodRequestName(method, RPCUtil.getImplClassName(interfaceClass))
                    .replace('.', '/');
            String className = RPCUtil.getProxyClassName(interfaceClass, method).replace('.', '/');
            ClassWriter cw = new ClassWriter(ClassWriter.COMPUTE_MAXS);
            cw.visit(V1_5, ACC_PUBLIC + ACC_SUPER, className, null, Type.getInternalName(RPCHandler.class), null);

            MethodVisitor mv = cw.visitMethod(ACC_PUBLIC, "<init>", "()V", null, null);
            mv.visitVarInsn(ALOAD, 0);
            mv.visitMethodInsn(INVOKESPECIAL, Type.getInternalName(RPCHandler.class), "<init>", "()V");
            mv.visitInsn(RETURN);
            mv.visitMaxs(1, 1);
            mv.visitEnd();

            mv = cw.visitMethod(ACC_PUBLIC, "handler",
                    "(Lcom/bigknow/appstore/common/rpc/RPCRequest;)Lcom/bigknow/appstore/common/rpc/RPCResponse;", null, null);
            mv.visitVarInsn(ALOAD, 0);
            mv.visitFieldInsn(GETFIELD, className, "impl", Type.getDescriptor(Object.class));
            mv.visitTypeInsn(CHECKCAST, Type.getInternalName(interfaceClass));
            Class<?>[] params = method.getParameterTypes();
            for (int i = 0; i < params.length; i++) {
                mv.visitVarInsn(ALOAD, 1);
                mv.visitTypeInsn(CHECKCAST, requestClass);
                Class<?> param = params[i];
                mv.visitMethodInsn(INVOKEVIRTUAL, requestClass, "getArg" + i, "()" + Type.getDescriptor(param));
            }
            mv.visitMethodInsn(INVOKEINTERFACE, Type.getInternalName(interfaceClass), method.getName(),
                    Type.getMethodDescriptor(method));
            Class<?> returnType = method.getReturnType();
            if (returnType == void.class) {
                mv.visitInsn(ACONST_NULL);
                mv.visitInsn(ARETURN);
                mv.visitMaxs(2, 1);
            } else {
                String responseInnerName = RPCUtil.getMethodResponseName(method,
                        RPCUtil.getImplClassName(interfaceClass)).replace('.', '/');
                int len = addStoreCode(mv, returnType, 2);
                mv.visitTypeInsn(NEW, responseInnerName);
                mv.visitInsn(DUP);
                mv.visitMethodInsn(INVOKESPECIAL, responseInnerName, "<init>", "()V");
                int responseIndex = 2 + len;
                mv.visitVarInsn(ASTORE, responseIndex);
                mv.visitVarInsn(ALOAD, responseIndex);
                addLoadCode(mv, returnType, 2);
                mv.visitMethodInsn(INVOKEVIRTUAL, responseInnerName, "setReturnObj",
                        "(" + Type.getDescriptor(returnType) + ")V");
                if (len == 1) {
                    mv.visitVarInsn(ALOAD, 3);
                } else {
                    mv.visitVarInsn(ALOAD, 4);
                }
                mv.visitInsn(ARETURN);
                mv.visitMaxs(3 + len, 3);
            }
            mv.visitEnd();
            cw.visitEnd();
            byte[] code = cw.toByteArray();
            try {
                Class<?> clz = RPCUtil.getClassLoader().defineClass(RPCUtil.getProxyClassName(interfaceClass, method),
                        code);
                RPCHandler methodHandler = (RPCHandler) clz.newInstance();
                methodHandler.setImpl(impl);
                methodHandlers.put(method.getName(), methodHandler);
            } catch (Throwable e) {
                throw new RuntimeException(e);
            }
        }
        RPCUtil.parseInterfaceClass(interfaceClass);
        rpcHandlerMap.put(interfaceClass.getName(), methodHandlers);
    }

    public <T> T getProxy(Class<T> interfaceClass, long timeout) {
        T proxy = getProxy(interfaceClass);
        RPCProxy temp = (RPCProxy) proxy;
        temp.setTimeout(timeout);
        return proxy;
    }

    @SuppressWarnings("unchecked")
    public <T> T getProxy(Class<T> interfaceClass) {
        try {
            Class<?> implClass = requestProxyMap.get(interfaceClass);
            if (implClass != null) {
                return (T) implClass.newInstance();
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        String className = RPCUtil.getImplClassName(interfaceClass);
        String internName = interfaceClass.getName().replace('.', '/');

        ClassWriter cw = new ClassWriter(ClassWriter.COMPUTE_MAXS);
        cw.visit(V1_5, ACC_PUBLIC + ACC_SUPER, className.replace('.', '/'), null, Type.getInternalName(RPCProxy.class),
                new String[]{internName});

        MethodVisitor mv = cw.visitMethod(ACC_PUBLIC, "<init>", "()V", null, null);
        mv.visitVarInsn(ALOAD, 0);
        mv.visitMethodInsn(INVOKESPECIAL, "com/bigknow/appstore/common/rpc/RPCProxy", "<init>", "()V");
        mv.visitInsn(RETURN);
        mv.visitMaxs(1, 1);
        mv.visitEnd();

        Method[] methods = interfaceClass.getMethods();
        for (Method method : methods) {
            createRequestClass(method, className, null);
            createResponseClass(method, className);
            createMethod(cw, method, className);
        }

        cw.visitEnd();
        RPCUtil.parseInterfaceClass(interfaceClass);

        byte[] code = cw.toByteArray();
        Class<T> implClass = (Class<T>) RPCUtil.getClassLoader().defineClass(className, code);
        try {
            requestProxyMap.put(interfaceClass, implClass);
            return (T) implClass.newInstance();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private void createRequestClass(Method method, String className, List<String> parmNames) {
        Class<?>[] parms = method.getParameterTypes();
        if (parms.length <= 0)
            return;
        ClassWriter cw = new ClassWriter(ClassWriter.COMPUTE_FRAMES);
        String requestClassName = RPCUtil.getMethodRequestName(method, className).replace('.', '/');
        cw.visit(V1_5, ACC_PUBLIC + ACC_SUPER, requestClassName, null, "com/bigknow/appstore/common/rpc/RPCRequest",
                new String[]{});
        MethodVisitor mv = cw.visitMethod(ACC_PUBLIC, "<init>", "()V", null, null);
        mv.visitVarInsn(ALOAD, 0);
        mv.visitMethodInsn(INVOKESPECIAL, "com/bigknow/appstore/common/rpc/RPCRequest", "<init>", "()V");
        mv.visitInsn(RETURN);
        mv.visitMaxs(1, 1);
        mv.visitEnd();

        java.lang.reflect.Type[] params = method.getGenericParameterTypes();
        Class<?>[] paramClasses = method.getParameterTypes();
        for (int i = 0; i < params.length; i++) {
            java.lang.reflect.Type param = params[i];
            cw.visitField(ACC_PRIVATE, "arg" + i, Type.getDescriptor(paramClasses[i]),
                    getSignature(param, paramClasses[i]), null);
            if (parmNames != null) {
                String paramName = parmNames.get(i);
                if (!paramName.equals("arg" + i)) {
                    cw.visitField(ACC_PRIVATE, paramName, Type.getDescriptor(paramClasses[i]),
                            getSignature(param, paramClasses[i]), null);
                }
            }
        }

        for (int i = 0; i < params.length; i++) {
            java.lang.reflect.Type param = params[i];
            String descriptor = Type.getDescriptor(paramClasses[i]);
            String signature = getSignature(param, paramClasses[i]);
            mv = cw.visitMethod(ACC_PUBLIC, "setArg" + i, "(" + descriptor + ")V", "(" + signature + ")V", null);
            mv.visitVarInsn(ALOAD, 0);
            addLoadCode(mv, paramClasses[i], 1);
            String defaultParamName = "arg" + i;
            String paramName = null;
            if (parmNames != null) {
                paramName = parmNames.get(i);
                if (paramName.equals(defaultParamName)) {
                    paramName = null;
                }
            }
            mv.visitFieldInsn(PUTFIELD, requestClassName, defaultParamName, descriptor);
            if (paramName != null) {
                mv.visitVarInsn(ALOAD, 0);
                addLoadCode(mv, paramClasses[i], 1);
                mv.visitFieldInsn(PUTFIELD, requestClassName, paramName, descriptor);
            }
            mv.visitInsn(RETURN);
            mv.visitMaxs(2, 1);
            mv.visitEnd();

            mv = cw.visitMethod(ACC_PUBLIC, "getArg" + i, "()" + descriptor, "()" + signature, null);
            mv.visitVarInsn(ALOAD, 0);
            mv.visitFieldInsn(GETFIELD, requestClassName, defaultParamName, descriptor);
            addReturnCode(mv, paramClasses[i]);
            mv.visitMaxs(1, 1);
            mv.visitEnd();

            if (paramName != null) {
                String uName = String.valueOf(paramName.charAt(0)).toUpperCase() + paramName.substring(1);
                mv = cw.visitMethod(ACC_PUBLIC, "set" + uName, "(" + descriptor + ")V", "(" + signature + ")V", null);
                mv.visitVarInsn(ALOAD, 0);
                addLoadCode(mv, paramClasses[i], 1);
                mv.visitFieldInsn(PUTFIELD, requestClassName, defaultParamName, descriptor);
                mv.visitVarInsn(ALOAD, 0);
                addLoadCode(mv, paramClasses[i], 1);
                mv.visitFieldInsn(PUTFIELD, requestClassName, paramName, descriptor);
                mv.visitInsn(RETURN);
                mv.visitMaxs(2, 1);
                mv.visitEnd();

                mv = cw.visitMethod(ACC_PUBLIC, "get" + uName, "()" + descriptor, "()" + signature, null);
                mv.visitVarInsn(ALOAD, 0);
                mv.visitFieldInsn(GETFIELD, requestClassName, defaultParamName, descriptor);
                addReturnCode(mv, paramClasses[i]);
                mv.visitMaxs(1, 1);
                mv.visitEnd();
            }
        }
        cw.visitEnd();
        byte[] code = cw.toByteArray();

        RPCUtil.getClassLoader().defineClass(RPCUtil.getMethodRequestName(method, className), code);
    }

    private String getSignature(java.lang.reflect.Type t, Class<?> clz) {
        if (t instanceof ParameterizedType) {
            ParameterizedType type = (ParameterizedType) t;
            String desc = Type.getDescriptor((Class<?>) type.getRawType());
            StringBuilder buf = new StringBuilder(desc.substring(0, desc.length() - 1));
            buf.append('<');
            for (java.lang.reflect.Type actualType : type.getActualTypeArguments()) {
                buf.append(getSignature(actualType, clz));
            }
            buf.append(">;");
            return buf.toString();
        } else if (t instanceof GenericArrayType) {
            return Type.getDescriptor(clz);
        } else if (t instanceof Class) {
            return Type.getDescriptor((Class<?>) t);
        } else {
            throw new UnsupportedOperationException("不支持的泛型类型(" + t + ")");
        }
    }

    private void createResponseClass(Method method, String className) {
        ClassWriter cw;
        MethodVisitor mv;
        byte[] code;
        Class<?> returnClz = method.getReturnType();
        java.lang.reflect.Type returnType = method.getGenericReturnType();
        if (returnClz != void.class) {
            cw = new ClassWriter(ClassWriter.COMPUTE_FRAMES);
            String responseClassName = RPCUtil.getMethodResponseName(method, className).replace('.', '/');
            cw.visit(V1_5, ACC_PUBLIC + ACC_SUPER, responseClassName, null, "com/bigknow/appstore/common/rpc/RPCResponse",
                    new String[]{});
            mv = cw.visitMethod(ACC_PUBLIC, "<init>", "()V", null, null);
            mv.visitVarInsn(ALOAD, 0);
            mv.visitMethodInsn(INVOKESPECIAL, "com/bigknow/appstore/common/rpc/RPCResponse", "<init>", "()V");
            mv.visitInsn(RETURN);
            mv.visitMaxs(1, 1);
            mv.visitEnd();

            String descriptor = Type.getDescriptor(returnClz);
            String signature = getSignature(returnType, returnClz);
            cw.visitField(ACC_PRIVATE, "returnObj", descriptor, signature, null);

            mv = cw.visitMethod(ACC_PUBLIC, "setReturnObj", "(" + descriptor + ")V", "(" + signature + ")V", null);
            mv.visitVarInsn(ALOAD, 0);
            addLoadCode(mv, returnClz, 1);
            mv.visitFieldInsn(PUTFIELD, responseClassName, "returnObj", descriptor);
            mv.visitInsn(RETURN);
            mv.visitMaxs(2, 1);
            mv.visitEnd();

            mv = cw.visitMethod(ACC_PUBLIC, "getReturnObj", "()" + descriptor, "()" + signature, null);
            mv.visitVarInsn(ALOAD, 0);
            mv.visitFieldInsn(GETFIELD, responseClassName, "returnObj", descriptor);
            addReturnCode(mv, returnClz);
            mv.visitMaxs(1, 1);
            mv.visitEnd();
            cw.visitEnd();
            code = cw.toByteArray();

            RPCUtil.getClassLoader().defineClass(RPCUtil.getMethodResponseName(method, className), code);
        }
    }

    private int addLoadCode(MethodVisitor mv, Class<?> param, int index) {
        if (param.isPrimitive()) {
            if (param == float.class) {
                mv.visitVarInsn(FLOAD, index);
            } else if (param == double.class) {
                mv.visitVarInsn(DLOAD, index);
                return 2;
            } else if (param == long.class) {
                mv.visitVarInsn(LLOAD, index);
                return 2;
            } else {
                mv.visitVarInsn(ILOAD, index);
            }
        } else {
            mv.visitVarInsn(ALOAD, index);
        }
        return 1;
    }

    private int addStoreCode(MethodVisitor mv, Class<?> param, int index) {
        if (param.isPrimitive()) {
            if (param == float.class) {
                mv.visitVarInsn(FSTORE, index);
            } else if (param == double.class) {
                mv.visitVarInsn(DSTORE, index);
                return 2;
            } else if (param == long.class) {
                mv.visitVarInsn(LSTORE, index);
                return 2;
            } else {
                mv.visitVarInsn(ISTORE, index);
            }
        } else {
            mv.visitVarInsn(ASTORE, index);
        }
        return 1;
    }

    private void addReturnCode(MethodVisitor mv, Class<?> param) {
        if (param.isPrimitive()) {
            if (param == float.class) {
                mv.visitInsn(FRETURN);
            } else if (param == double.class) {
                mv.visitInsn(DRETURN);
            } else if (param == long.class) {
                mv.visitInsn(LRETURN);
            } else {
                mv.visitInsn(IRETURN);
            }
        } else {
            mv.visitInsn(ARETURN);
        }
    }

    private void createMethod(ClassWriter cw, Method method, String className) {
        MethodVisitor mv = cw.visitMethod(ACC_PUBLIC, method.getName(), Type.getMethodDescriptor(method), null, null);
        mv.visitCode();
        String requestClzName = RPCUtil.getMethodRequestName(method, className).replace('.', '/');

        mv.visitTypeInsn(NEW, requestClzName);
        mv.visitInsn(DUP);
        mv.visitMethodInsn(INVOKESPECIAL, requestClzName, "<init>", "()V");

        Class<?>[] params = method.getParameterTypes();
        int paramsLength = 0;
        for (Class<?> param : params) {
            if (param == long.class || param == double.class) {
                paramsLength += 2;
            } else {
                paramsLength++;
            }
        }
        int requestVarIndex = paramsLength + 1;
        mv.visitVarInsn(ASTORE, requestVarIndex);
        int index = 1;
        for (int i = 0; i < params.length; i++) {
            mv.visitVarInsn(ALOAD, requestVarIndex);
            int len = addLoadCode(mv, params[i], index);
            mv.visitMethodInsn(INVOKEVIRTUAL, requestClzName, "setArg" + i, "(" + Type.getDescriptor(params[i]) + ")V");
            index += len;
        }

        mv.visitVarInsn(ALOAD, requestVarIndex);
        mv.visitLdcInsn(className.substring(0, className.length() - 5));
        mv.visitMethodInsn(INVOKEVIRTUAL, requestClzName, "setInterfaceName", "(Ljava/lang/String;)V");

        mv.visitVarInsn(ALOAD, requestVarIndex);
        mv.visitLdcInsn(method.getName());
        mv.visitMethodInsn(INVOKEVIRTUAL, requestClzName, "setMethodName", "(Ljava/lang/String;)V");

        mv.visitVarInsn(ALOAD, requestVarIndex);
        mv.visitVarInsn(ALOAD, 0);
        mv.visitFieldInsn(GETFIELD, "com/bigknow/appstore/common/rpc/RPCProxy", "timeout", Type.getDescriptor(long.class));
        mv.visitMethodInsn(INVOKESTATIC, Type.getInternalName(RPCClient.class), "invoke",
                "(Lcom/bigknow/appstore/common/rpc/RPCRequest;J)Lcom/bigknow/appstore/common/rpc/RPCResponse;");
        mv.visitVarInsn(ASTORE, paramsLength + 2);

        mv.visitVarInsn(ALOAD, paramsLength + 2);
        mv.visitMethodInsn(INVOKEVIRTUAL, Type.getInternalName(RPCResponse.class), "isError", "()Z");
        Label label = new Label();
        mv.visitJumpInsn(IFEQ, label);

        mv.visitTypeInsn(NEW, Type.getInternalName(RuntimeException.class));
        mv.visitInsn(DUP);
        mv.visitVarInsn(ALOAD, paramsLength + 2);
        mv.visitMethodInsn(INVOKEVIRTUAL, Type.getInternalName(RPCResponse.class), "getErrorMsg",
                "()Ljava/lang/String;");
        mv.visitMethodInsn(INVOKESPECIAL, Type.getInternalName(RuntimeException.class), "<init>",
                "(Ljava/lang/String;)V");
        mv.visitInsn(ATHROW);

        mv.visitLabel(label);
        Class<?> returnType = method.getReturnType();
        String responseClzName = RPCUtil.getMethodResponseName(method, className).replace('.', '/');
        if (returnType != void.class) {
            mv.visitVarInsn(ALOAD, paramsLength + 2);
            mv.visitTypeInsn(CHECKCAST, responseClzName);
            mv.visitMethodInsn(INVOKEVIRTUAL, responseClzName, "getReturnObj", "()" + Type.getDescriptor(returnType));
            addReturnCode(mv, returnType);
        } else {
            mv.visitInsn(RETURN);
        }
        mv.visitMaxs(paramsLength + 3, params.length + 2);
        mv.visitEnd();
    }

    public static RPCResponse invoke(final RPCRequest req, long timeout) {
        String msgId = "S_" + idGen.incrementAndGet();
        RPCResponse response = null;
        responseMap.put(msgId, null);
        String lock = msgId.intern();
        timeout = timeout == 0 ? 30000 : timeout;
        synchronized (lock) {
            writeData(msgId, req);
            try {
                lock.wait(timeout);
            } catch (InterruptedException e) {
            }
            response = responseMap.remove(msgId);
        }
        if (response == null) {
            response = new RPCResponse();
            response.setError(true);
            response.setErrorMsg("请求超时");
        }
        return response;
    }

    class Reader extends Thread {
        public void run() {
            while (!closed && connected) {
                try {
                    final String msgId = readString();
                    final String clzName = readString();
                    final String dataStr = readString();
                    pool.submit(new Runnable() {
                        public void run() {
                            handler(msgId, clzName, dataStr);
                        }
                    });
                } catch (Throwable e) {
                    connected = false;
//                    System.out.println(String.format("服务器连接已断开（%s:%s）", serviceHost, servicePort));
                    if (!closed) {
                        createReconnetTask();
                    }
                }
            }
        }
    }

    private synchronized static void writeData(String msgId, Object data) {
        for (int i = 0; !connected && i < 50; i++) {
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
            }
        }
        try {
            String json = JSONValue.toJSONString(data, JSONStyle.NO_COMPRESS);
            writeStr(msgId);
            writeStr(data.getClass().getName());
            writeStr(json);
            dataOutput.flush();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private String readString() throws IOException {
        int len = dataInput.readInt();
        if (len <= 0 || len > 8 * 1024 * 1024) {
            IOException e = new IOException("协议错误，数据长度：" + len);
            throw e;
        }
        byte[] data = new byte[len];
        dataInput.readFully(data);
        if (ZipUtil.isCompress(data)) {
            data = ZipUtil.uncompress(data);
        }
        return new String(data, "UTF-8");
    }

    public static void writeStr(String str) throws IOException {
        byte[] data = str.getBytes("UTF-8");
        if (data.length >= Constants.MIN_COMPRESS_LENGTH) {
            data = ZipUtil.compress(data);
            dataOutput.writeInt(data.length + RPCConstants.compressHeader.length);
            dataOutput.write(RPCConstants.compressHeader);
            dataOutput.write(data);
        } else {
            dataOutput.writeInt(data.length);
            dataOutput.write(data);
        }
    }

    @SuppressWarnings("unchecked")
    private void handler(String msgId, String clzName, String data) {
        Class<?> clz = null;
        Object msg = null;
        try {
            clz = RPCUtil.getClassLoader().loadClass(clzName);
            msg = JSONValue.parse(data, clz);
            if (msg instanceof RPCRequest) {
                RPCRequest req = (RPCRequest) msg;
                Map<String, RPCHandler> handlerMap = rpcHandlerMap.get(req.getInterfaceName());
                if (handlerMap == null) {
                    returnErrorMsg("不能处理的请求(" + req.getInterfaceName() + "::" + req.getMethodName() + ")", msgId);
                } else {
                    RPCHandler handler = handlerMap.get(req.getMethodName());
                    if (handler == null) {
                        returnErrorMsg("不能处理的请求(" + req.getInterfaceName() + "::" + req.getMethodName() + ")", msgId);
                    } else {
                        try {
                            RPCResponse response = handler.handler((RPCRequest) msg);
                            if (response != null) {
                                writeData(msgId, response);
                            } else {
                                response = new RPCResponse();
                                writeData(msgId, response);
                            }
                        } catch (Throwable t) {
                            returnErrorMsg(
                                    "请求处理错误(" + req.getInterfaceName() + "::" + req.getMethodName() + ","
                                            + getRootMessage(t) + ")", msgId);
                        }
                    }
                }
            } else if (msg instanceof RPCResponse) {
                RPCResponse response = (RPCResponse) msg;
                String lock = msgId.intern();
                synchronized (lock) {
                    if (responseMap.containsKey(msgId)) {
                        responseMap.put(msgId, response);
                        lock.notify();
                    }
                }
            }
        } catch (Throwable t) {
            connected = false;
            if (!closed) {
                createReconnetTask();
            }
            t.printStackTrace();
        }
    }

    private String getRootMessage(Throwable t) {
        Throwable cause = t.getCause();
        while (cause != null && cause != t) {
            t = cause;
            cause = t.getCause();
        }
        return t.getMessage();
    }

    private void returnErrorMsg(String msg, String msgId) {
        RPCResponse response = new RPCResponse();
        response.setError(true);
        response.setErrorMsg(msg);
        writeData(msgId, response);
    }
}